package com.aispeech.asr.apis.ssc.beans;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.function.Function;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.core.io.Resource;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;

import com.aispeech.asr.apis.ssc.constants.Progress;
import com.aispeech.asr.apis.ssc.data.domain.AudioInfo;
import com.aispeech.asr.apis.ssc.data.domain.TaskDoc;
import com.aispeech.asr.apis.ssc.event.SscEvent;
import com.aispeech.asr.apis.ssc.service.ProgressManager;
import com.aispeech.asr.comm.enums.ErrorCode;
import com.aispeech.asr.comm.exceptions.AsrException;
import com.aispeech.asr.comm.utils.FileUtils;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;

@Component
@Slf4j
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class Downloader implements Function<SscEvent, Mono<SscEvent>> {
    @Autowired
    WebClient webClient;

    @Autowired
    MongoOperations mongoOperations;
    @Autowired
    ProgressManager progressManager;
    
    @SneakyThrows
    @Override
    public Mono<SscEvent> apply(SscEvent event) {
        Mono<SscEvent> mono = Mono.just(event);
        TaskDoc taskDoc = event.getTaskDoc();
        log.info("[ Downloader ] start。。。taskId:{}",taskDoc.getId());
        if (taskDoc.getProgress() >= Progress.DOWNLOADED) return mono;  // 不要重复下载已成功的任务
        progressManager.syncToMongo(taskDoc, Progress.DOWNLOADING);
        taskDoc.addMetric("downloading_t");

        return mono
                .flatMap(this::downloadAudio)
                .doOnNext(ctx -> {
                    progressManager.syncToMongo(taskDoc, Progress.DOWNLOADED);
                    taskDoc.addMetric("download_t");
                });
    }

    private Mono<SscEvent> downloadAudio(SscEvent event) {
        TaskDoc taskDoc = event.getTaskDoc();
        String audioUrl = (String) taskDoc.getRequest().getOrDefault("filePath",
                taskDoc.getRequest().get("audioFilePath"));
        log.info("[ Downloader ] url: {}", audioUrl);
        
        if (StringUtils.isNotEmpty(audioUrl)) {
        	try {
        		URL url = new URL(audioUrl);
        		log.info("[ Downloader ] url.toURI(): {}", url.toURI());
        		return webClient.get()
                        .uri(url.toURI())
                        .retrieve()
                        .bodyToMono(Resource.class)
                        .map(resource -> prepareTaskWorkDir(event, taskDoc, resource));
        	}catch(Exception e) {
        		return Mono.error(new AsrException(ErrorCode.ERR_AUDIO_DOWNLOAD_FAILED.errid, "url: " + audioUrl));
        	}
            
        } else {
            return Mono.error(new AsrException(ErrorCode.ERR_AUDIO_DOWNLOAD_FAILED.errid, "url: " + audioUrl));
        }
    }

    @SneakyThrows
    private SscEvent prepareTaskWorkDir(SscEvent event, TaskDoc taskDoc, Resource resource) {
        AudioInfo audioInfo = event.getAudioInfo();
        log.info("[ Downloader ] downloaded. task: {}, audio length: {}", taskDoc.getId(), resource.contentLength());
        try {
            Path path = Paths.get(audioInfo.getWorkDir());

            if (path.toFile().exists()) {
                FileUtils.delete(path, true);
            }

            Files.createDirectories(path);
            Files.copy(resource.getInputStream(), Paths.get(audioInfo.getOrignAudio()));

            progressManager.syncToMongo(taskDoc, Progress.DOWNLOADED);
            
            log.info("[ Downloader ] end。。。taskId:{}",taskDoc.getId());
            return event;
        } catch (IOException e) {
            throw new AsrException(ErrorCode.ERR_AUDIO_DOWNLOAD_FAILED.errid, e.getMessage());
        }
    }
}
